package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.scalar.NotBlank;
import com.sui.bigdata.flink.sql.function.util.PhoneHHIAccum;

import org.apache.flink.table.functions.AggregateFunction;

import java.util.Iterator;

public class UserknPhoneUnusualNum extends AggregateFunction<Integer, PhoneHHIAccum> {

    @Override
    public PhoneHHIAccum createAccumulator() {
        return new PhoneHHIAccum();
    }

    @Override
    public Integer getValue(PhoneHHIAccum phoneHHIAccum) {
        return phoneHHIAccum.getCount();
    }

    public void accumulate(PhoneHHIAccum phoneHHIAccum, String phone) {
        if (NotBlank.eval(phone) && !phoneHHIAccum.exitsPhone(phone) && phone.contains("<%!-Non-regularized-%>")) {
            phoneHHIAccum.addphone(phone);
            phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
        }
    }

    public void merge(PhoneHHIAccum phoneHHIAccum, Iterable<PhoneHHIAccum> it) {
        Iterator<PhoneHHIAccum> iter = it.iterator();
        while (iter.hasNext()) {
            PhoneHHIAccum a = iter.next();
            phoneHHIAccum.setCount(a.getCount()+phoneHHIAccum.getCount());
            phoneHHIAccum.getRepetition().addAll(a.getRepetition());
        }
    }

}
